package report

import java.util.Properties

import Configer.Configer
import com.google.gson.Gson
import org.apache.commons.lang.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}
import scalikejdbc.{DB, SQL}
import utils.KPITotal

//媒体分析广播变量
object AppAnalysis {
  def main(args: Array[String]): Unit = {
    //sparkcontext
    val conf = new SparkConf().setAppName(s"${this.getClass.getName}")
      .setMaster("local[*]").set("spark.serializer",Configer.serializer)
    val sc = new SparkContext(conf)
    //sqlContext
    val sQLContext = new SQLContext(sc)
    //读取数据
    val dictLines = sc.textFile("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\app_dict.txt")
    val dataFrame = sQLContext.read.parquet("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\parquet")
    //广播变量
    val dictMap: collection.Map[String, String] = dictLines.map(line => line.split("\t", line.length)).filter(_.length >= 5).map(arr => (arr(4), arr(1)))
      .collectAsMap()
    val dictBro = sc.broadcast(dictMap)
    //分析数据
    val result: RDD[(String, List[Double])] = dataFrame.map(row => {
      var appName = row.getAs[String]("appname")
      val appId = row.getAs[String]("appid")
      if (StringUtils.isEmpty(appName)) {
        if (StringUtils.isEmpty(appId)) {
          appName = "某app"
        } else {
          val appName = dictBro.value.getOrElse(appId, appId)
        }
      }
      (appName, KPITotal.KPI(row))
    }).reduceByKey((list1, list2) => list1 zip list2 map (li => li._1 + li._2))
    //转换成df
    import sQLContext.implicits._
    val frame = result.map(arr => (arr._1, arr._2(0), arr._2(1), arr._2(2), arr._2(3), arr._2(4), arr._2(7), arr._2(8), arr._2(5), arr._2(6)))
      .toDF("appName", "allreq", "effreq", "adreq", "adjoin", "adsucess", "adshow", "adclick", "adpay", "adwin")
    //存储数据
//    val props = new Properties()
//    props.setProperty("user",Configer.user)
//    props.setProperty("password",Configer.password)
//    props.setProperty("driver",Configer.driver)
//    frame.write.mode(SaveMode.Overwrite).jdbc(Configer.url,"Area34",props)
    result.mapPartitions(partition=>{
      DB.localTx{implicit session=>
        partition.map(arr=>{
          SQL("insert into Area34 values (?,?,?,?)").bind(arr._1,arr._2(0),"1","2")
            .update().apply()
        })
      }
    })
    //frame.write.json()
//    result.map(arr=>{
//      val gson = new Gson()
//      gson.toJson(arr._1)
//    })

    //释放资源
    sc.stop()
  }
}
